package org.example.api.process;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.example.api.bean.SensorReading;

/**
 * @author huangqihan
 * @date 2021/2/25
 */
public class SideOutput {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String host = "localhost";
        int port = 7777;

        // Use NetCat to read the data source. nc -lp 7777
        DataStreamSource<String> inputStream = env.socketTextStream(host, port);

        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("low") {
        };

        SingleOutputStreamOperator<Object> high = dataStream
                .process(new ProcessFunction<SensorReading, Object>() {
                    @Override
                    public void processElement(SensorReading sensorReading, Context context, Collector<Object> collector) throws Exception {
                        if (sensorReading.getTemperature() > 30) {
                            collector.collect(sensorReading);
                        }else {
                            context.output(outputTag,sensorReading);
                        }
                    }
                });

        high.print();
        DataStream<SensorReading> low = high.getSideOutput(outputTag);
        low.print();
        env.execute();
    }
}
